package com.yz.kafka.consumers.offsets;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import javax.annotation.Resource;
import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.Collections;
import java.util.Properties;

/**
 * Kafka消费者——异步提交
 *
 * @author yuanzheng
 * @since 2021/3/7
 */
public class SimpleConsumerAsyn {
    @Resource
    private static final String topic = "Hello-Kafka";

    @Resource
    private static final String group = "group1";

    @Resource
    private static final String brokerList = "group1";



    public static void main(String[] args) {


        Properties props = new Properties();
        props.put("bootstrap.servers", brokerList);
        /*指定分组 ID*/
        props.put("group.id", group);
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        /*订阅主题 (s)*/
        consumer.subscribe(Collections.singletonList(topic));
        try {
            while (true) {
                /*异步提交并定义回调*/
                ConsumerRecords<String, String> records = consumer.poll(Duration.of(100, ChronoUnit.MILLIS));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("topic = %s,partition = %d, key = %s, value = %s, offset = % d,\n ", record.topic(), record.partition(), record.key(), record.value(), record.offset());
                }
                consumer.commitAsync((offsets, exception) -> {
                    if (exception != null) {
                        System.out.println("错误处理");
                    } else {
                        offsets.forEach((x, y) -> System.out.printf("topic = %s,partition = %d, offset = %s", x.topic(), x.partition(), y.offset()));
                    }
                });
            }
        } finally {
            consumer.close();
        }
    }
}
